#include <cinttypes>
#include <cstdint>
#include <list>
#include <cmath>
#include <opus.h>
#include <string.h>
#include <string>
#include <stdio.h>
#include <stdlib.h>
#include <netinet/in.h>

#include "audio_recveq.h"
#include "codecs.hpp"
#include "remote_mixer.hpp"
#include "rtp.hpp"
#include "rtp/rtp_exten.hpp"
#include "xcutil.h"
#include "xm_app_config.h"
#include "xm_log.h"
#include "xmixer.h"
#include "xrtc_endpoint.h"
#include "xtlv_file.h"

#define mix_dbgi dbgi
#define mix_dbge dbge

//         
//          +----------+    +----------+    +----------+    +----------+
//          |  talker  |    |  talker  |    | passive  |    |candidate |
//          +----------+    +----------+    +----------+    +----------+
//              |                  |           |                 |
//              |                  |           |                 v
//              |                  |           |            +------------+
//              |                  |           |            |            |
//              |                  |           |            |   next     |
//              v                  v           v            |   passive  |
//          +---------+         +----------------+          |            |
//          |         |         |                |          +------------+
//          |   mix   | <-------+  fade in&out   |<---+
//          |         |         |                |    |
//          +---------+         +----------------+    |     +------------+
//              |                                     +-----|            |
//              |                                           |   last     |
//              +-------------------------------+---------->|   frame    |
//              |                |              |           |            |
//              v                v              |           +------------+
//          +---------+     +----------+    +----------+
//          |         |     |          |    |          |
//          |   sub   |     |   sub    |    | broadcast|
//          |         |     |          |    | encode   |
//          +---------+     +----------+    |          |
//               |               |          +----------+
//               v               v               |  |
//          +----------+    +----------+         |  +----------------------------+
//          |  encode  |    |  encode  |         |              |                |
//          +----------+    +----------+         |              |                |
//               |               |               |              |                |
//               v               v               v              v                v
//          +----------+    +----------+    +----------+    +----------+    +----------+
//          |  talker  |    |  talker  |    | passive  |    |candidate |    | audience |
//          +----------+    +----------+    +----------+    +----------+    +----------+
//         



#define ZERO_ALLOC(o, type, sz) do{o=(type)malloc(sz); memset(o, 0, sz);}while(0)



#define MIX_AUDIO_CHANNELS 2
#define MIX_AUIDO_FS 16000
#define MIX_AUDIO_SAMPLES_20MS (2*(MIX_AUIDO_FS/100)*MIX_AUDIO_CHANNELS)
#define MIX_AUIDO_MAX_TALKER 5
#define MIX_RTP_SSRC 754039198
#define MIX_RTP_PT 111
#define MIX_AUDIO_RESET_LEVEL (-127)
#define MIX_AUDIO_MIN_LEVEL (-127)
#define MIX_AUDIO_TALKER_SILENCE_PERIOD 1500
#define MIX_AUDIO_CONTRIBUTING_LEVEL (-50)

#define MIX_AUDIO_RTP_HEADER_MAX_SIZE 128
#define MIX_AUDIO_RTP_PAYLOAD_MAX_SIZE 1600
#define MIX_AUDIO_OPUS_APP OPUS_APPLICATION_AUDIO
#define MIX_AUDIO_HISTORY_SIZE (2000 / 20)

typedef struct xmixer_stream_st* xmixer_stream_t;
typedef struct xmixer_talker_st* xmixer_talker_t;

// debug recording
//
int open_debug_output_tlv(xmixer_t mixer, xmixer_stream_t mxstream);
int close_debug_output_tlv(xmixer_stream_t mxstream);
int write_debug_output_tlv(xmixer_stream_t mxstream, int encoder_id, uint8_t* buf, size_t len);
int open_debug_input_tlv(xmixer_t mixer, xmixer_stream_t mxstream);
int close_debug_input_tlv(xmixer_stream_t mxstream);
int write_debug_input_tlv(xmixer_stream_t mxstream, int encoder_id, uint8_t* buf, size_t len);

// static void mixer_remote_stream_timeout(void* data, rmixer_stream* stream);
// static void mixer_remote_stream_new(void* data, rmixer_stream* stream);
// static void mixer_remote_stream_data(void* data, rmixer_stream* stream, unsigned char* buf, int size);

struct xmixer_stream_st {
    bool is_remote;
    uint32_t ssrc;

    rtc_endpoint_t rtc;
    char* rtc_id;

    xmixer_t mixer;
    std::list<xmixer_stream_t>::iterator it_for_owner;
    xmixer_talker_t talker;  // NULL if NOT talker
    audio_recveq_t recveq ;
    uint16_t rtp_seq;
    OpusEncoder * opus_enc;
    bool in_dtx; // set to true when sending first one of dtx packets
    bool need_warmup;
    int last_audio_level;
    int audio_level_sum;
    int num_audio_levels;
    
    uint8_t encoded_buf[MIX_AUDIO_RTP_HEADER_MAX_SIZE+MIX_AUDIO_RTP_PAYLOAD_MAX_SIZE];

    // debug recording
    int64_t debug_output_ts;
    tlv_file_t debug_output_tlv;
    tlv_file_t debug_input_tlv;
    // int opus_err;
};

typedef enum{
    TALK_OP_NONE = 0,
    TALK_OP_RAMPIN,
    TALK_OP_NORMAL,
    TALK_OP_RAMPOUT,
}talker_operation_t;

struct xmixer_talker_st{
    xmixer_stream_t mxstream;
    int16_t pcm_buf[MIX_AUDIO_SAMPLES_20MS];
    int pcm_samples;
    talker_operation_t op;
    int8_t audio_level;
    int64_t silence_start_time;
    double audio_level_hist[MIX_AUDIO_HISTORY_SIZE];
    int audio_level_hist_h;
    int audio_level_hist_t;
    double audio_level_avg;
};

class mixer_debug;
mixer_debug* open_mixer_debug_file(xmixer_t mixer);
void close_mixer_debug_file(mixer_debug* debug);
void write_mixer_debug_file(mixer_debug* debug, rtc_endpoint_t rtc, int16_t* samples, int samples_nb);
void flush_mixer_debug_file(mixer_debug* debug);

struct xmixer_st{
    char id[64];
    int is_loopback;
    std::list<xmixer_stream_t> * mxstreams;
    std::map<uint32_t, xmixer_stream_t> streams;
    struct xmixer_talker_st talkers[MIX_AUIDO_MAX_TALKER];
    struct xmixer_talker_st passive_talker;
    xmixer_stream_t candidate_stream;
    gint64 next_mix_time; // in micro-seconds
    gint64 num_mixed_frames;
    OpusEncoder * opus_enc_main;
    int16_t stream_frame[MIX_AUDIO_SAMPLES_20MS]; // 3840
    int16_t mix_frame[MIX_AUDIO_SAMPLES_20MS];
    bool has_prev_mix_frame;
    int16_t prev_mix_frame[MIX_AUDIO_SAMPLES_20MS];
    uint8_t mixed_encoded_buf[MIX_AUDIO_RTP_HEADER_MAX_SIZE+MIX_AUDIO_RTP_PAYLOAD_MAX_SIZE];
    int candidate_audio_level;
    int64_t talker_update_seq;
    int64_t last_dump_time;
    int64_t start_time;
    char* recording_tlv_pathname;
    tlv_file_t recording_tlv;
    uint16_t recording_rtp_seq;
    bool debug_enabled;
    mixer_debug* debug;
    int max_talkers;

    rmixer* remote;
};

class csrc_audio_level {
public:
    uint32_t csrc;
    int8_t audio_level;
};

#define MAX_ENERGY 4294967295UL
// Linear ramping over 80 samples.
// TODO(hellner): ramp using fix point?
static const float rampArray[] = {0.0000f, 0.0127f, 0.0253f, 0.0380f,
    0.0506f, 0.0633f, 0.0759f, 0.0886f,
    0.1013f, 0.1139f, 0.1266f, 0.1392f,
    0.1519f, 0.1646f, 0.1772f, 0.1899f,
    0.2025f, 0.2152f, 0.2278f, 0.2405f,
    0.2532f, 0.2658f, 0.2785f, 0.2911f,
    0.3038f, 0.3165f, 0.3291f, 0.3418f,
    0.3544f, 0.3671f, 0.3797f, 0.3924f,
    0.4051f, 0.4177f, 0.4304f, 0.4430f,
    0.4557f, 0.4684f, 0.4810f, 0.4937f,
    0.5063f, 0.5190f, 0.5316f, 0.5443f,
    0.5570f, 0.5696f, 0.5823f, 0.5949f,
    0.6076f, 0.6203f, 0.6329f, 0.6456f,
    0.6582f, 0.6709f, 0.6835f, 0.6962f,
    0.7089f, 0.7215f, 0.7342f, 0.7468f,
    0.7595f, 0.7722f, 0.7848f, 0.7975f,
    0.8101f, 0.8228f, 0.8354f, 0.8481f,
    0.8608f, 0.8734f, 0.8861f, 0.8987f,
    0.9114f, 0.9241f, 0.9367f, 0.9494f,
    0.9620f, 0.9747f, 0.9873f, 1.0000f};

static const int rampSize = sizeof(rampArray)/sizeof(rampArray[0]);

static void xmixer_handle_remote_stream_new(void* data, uint32_t ssrc);
static void xmixer_handle_remote_stream_data(void* data, uint32_t ssrc, const unsigned char* buf, int size);
static void xmixer_handle_remote_stream_destroy(void* data, uint32_t ssrc);

auto g_rmixer_callback = rmixer_callback {
        .on_stream_destroy = xmixer_handle_remote_stream_destroy,
        .on_stream_new = xmixer_handle_remote_stream_new,
        .on_stream_data = xmixer_handle_remote_stream_data,
};

// calculate audio level according to rfc6465
// see https://tools.ietf.org/html/rfc6465#appendix-A.1 for details
// returns -127 ~ 0
static inline
std::pair<int8_t,double> calculate_audio_level(int16_t *samples, size_t n) {
    if (n == 0) {
        return std::pair<int8_t, double>(0, 0.0);
    }
    double energy = 0;
    double rms = 0;
    for(size_t i = 0; i < n; ++i){
        double v = static_cast<double>(samples[i]);
        v /= 32767;
        rms += v * v;
        energy += v * v;
    }
    rms = std::sqrt(rms / n);
    energy = energy / n;
    int db = 20 * std::log10(rms);
    if (db < -127) {
        db = -127;
    } else if (db > 0) {
        db = 0;
    }
    return std::pair<int8_t, double>(db, energy);
}

static inline
void rampin_talker_pcm(xmixer_talker_t talker){
    if(talker->pcm_samples < (rampSize*MIX_AUDIO_CHANNELS)){
        return ;
    }
    
    for(int i = 0; i < rampSize; i++){
        float w = rampArray[i];
        for(int ch = 0; ch < MIX_AUDIO_CHANNELS; ch++){
            int pcm_pos = i * MIX_AUDIO_CHANNELS + ch;
            talker->pcm_buf[pcm_pos] = static_cast<int16_t>(w * talker->pcm_buf[pcm_pos]);
        }
    }
}

static inline
void rampout_talker_pcm(xmixer_talker_t talker){
    if(talker->pcm_samples < (rampSize*MIX_AUDIO_CHANNELS)){
        if(talker->pcm_samples > 0){
            dbge("ERROR: should NOT happen, diff ramp size");
            memset(&talker->pcm_buf[0], 0,  talker->pcm_samples * sizeof(int16_t));
        }
        return ;
    }
    
    for(int i = 0; i < rampSize; i++){
        float w = rampArray[rampSize - 1 - i];
        for(int ch = 0; ch < MIX_AUDIO_CHANNELS; ch++){
            int pcm_pos = i * MIX_AUDIO_CHANNELS + ch;
            talker->pcm_buf[pcm_pos] = static_cast<int16_t>(w * talker->pcm_buf[pcm_pos]);
        }
    }
    
    memset(&talker->pcm_buf[rampSize*MIX_AUDIO_CHANNELS], 0,  (talker->pcm_samples - rampSize*MIX_AUDIO_CHANNELS) * sizeof(int16_t));
}

static inline
void bind_stream_with_talker(xmixer_talker_t talker, xmixer_stream_t mxstream){
    if(talker){
        if(talker->mxstream != NULL && talker->mxstream != mxstream){
            dbge("bind non-empty talker !!!");
        }
        talker->mxstream = mxstream;
    }
    if(mxstream){
        if(mxstream->talker != NULL && mxstream->talker != talker){
            dbge("bind non-empty mxstream !!!");
        }
        mxstream->talker = talker;
    }
}

static inline
void unbind_stream_from_talker(xmixer_talker_t talker){
    if(talker){
        if(talker->mxstream){
            talker->mxstream->talker = NULL;
            talker->mxstream = NULL;
        }
    }
}

static inline
void dump_talkers(xmixer_t mixer){
    char str[1024];
    int strlen = 0;
    int64_t now = get_timestamp_ms();
    int elapsed = int(now - mixer->start_time);
    strlen += snprintf(str+strlen, sizeof(str) - strlen, "%04d.%03d", elapsed / 1000, elapsed % 1000);
    strlen += snprintf(str+strlen, sizeof(str) - strlen, " active ");
    for(int ti = 0; ti < mixer->max_talkers; ti++){
        xmixer_talker_t talker = &mixer->talkers[ti];
        xmixer_stream_t mxstream = talker->mxstream;
        
        if(!mxstream){
            strlen += snprintf(str+strlen, sizeof(str) - strlen, "[NULL] ");
        } else {
            strlen += snprintf(str+strlen, sizeof(str) - strlen, "[%s/%u/%fdBov] ", mxstream->rtc_id, mxstream->ssrc, talker->audio_level_avg);
        }
    }

    strlen += snprintf(str+strlen, sizeof(str) - strlen, "passive ");
    xmixer_stream_t mxstream = mixer->passive_talker.mxstream;
    if(!mxstream){
        strlen += snprintf(str+strlen, sizeof(str) - strlen, "[NULL] ");
    } else {
        strlen += snprintf(str+strlen, sizeof(str) - strlen, "[%s/%u/%ddBov] ", mxstream->rtc_id, mxstream->ssrc, mixer->passive_talker.audio_level);
    }

    strlen += snprintf(str+strlen, sizeof(str) - strlen, "cand ");
    if (mixer->candidate_stream == nullptr) {
        strlen += snprintf(str+strlen, sizeof(str) - strlen, "[NULL] ");
    } else {
        xmixer_stream_t mxstream = mixer->candidate_stream;
        strlen += snprintf(str+strlen, sizeof(str) - strlen, "[%s/%u/%ddBov] ", mxstream->rtc_id, mxstream->ssrc, mxstream->last_audio_level);
    }
    mix_dbgi("mixer |%s|: %s", mixer->id, str);
}

void xmixer_stream_delete(xmixer_stream_t mxstream) {
    if(!mxstream){
        return ;
    }
    
    xmixer_t mixer = mxstream->mixer;
    mix_dbgi("mixer |%s| delete rtc |%s|", mixer->id, mxstream->rtc_id);
    mixer->mxstreams->erase(mxstream->it_for_owner);

    if(mxstream->talker){
        unbind_stream_from_talker(mxstream->talker);
        ++mixer->talker_update_seq;
        dump_talkers(mixer);
    }
    
    if(mxstream == mixer->candidate_stream){
        mixer->candidate_stream = NULL;
    }

    if(mixer->mxstreams->size() == 0){
        mixer->next_mix_time = LLONG_MAX;
        dbgi("exist mxstream 0, stop mixing");
    }
    
    if(mxstream->recveq){
        audio_recveq_delete(mxstream->recveq);
        mxstream->recveq = NULL;
    }
    
    if(mxstream->opus_enc){
        opus_encoder_destroy(mxstream->opus_enc);
        mxstream->opus_enc = NULL;
    }

    if (mxstream->debug_output_tlv) {
        close_debug_output_tlv(mxstream);
    }

    if (mxstream->debug_input_tlv) {
        close_debug_input_tlv(mxstream);
    }

    if (mxstream->rtc_id) {
        g_free(mxstream->rtc_id);
    }
    free(mxstream);
}

void xmixer_delete(xmixer_t mixer){
    if(!mixer) return ;
    mix_dbgi("mixer |%s| deleted", mixer->id);
    if(mixer->mxstreams){
        while(!mixer->mxstreams->empty()){
            xmixer_stream_delete(mixer->mxstreams->front());
        }
        delete mixer->mxstreams;
        mixer->mxstreams = NULL;
    }
    
    if(mixer->opus_enc_main){
        opus_encoder_destroy(mixer->opus_enc_main);
        mixer->opus_enc_main = NULL;
    }

    if (mixer->recording_tlv_pathname) {
        free(mixer->recording_tlv_pathname);
        mixer->recording_tlv_pathname = nullptr;
    }

    if (mixer->recording_tlv) {
        tlv_file_close(mixer->recording_tlv);
        mixer->recording_tlv = nullptr;
    }

    if (mixer->debug) {
        close_mixer_debug_file(mixer->debug);
        mixer->debug = nullptr;
    }

    if (mixer->remote) {
        rmixer_del(mixer->remote);
    }

    //free(mixer);
    delete(mixer);
}

xmixer_t xmixer_create(const char* id, int is_loopback, bool debug_enabled, std::string& addr){
    int ret = -1;
    xmixer_t mixer = NULL;
    do {
        //ZERO_ALLOC(mixer, xmixer_t, sizeof(struct xmixer_st));
        mixer = new xmixer_st();
        memcpy(mixer->id, id, strlen(id));
        mixer->is_loopback = is_loopback;
        mixer->next_mix_time = LLONG_MAX;
        mixer->mxstreams = new std::list<xmixer_stream_t>();
        mixer->candidate_audio_level = MIX_AUDIO_RESET_LEVEL;
        mixer->debug_enabled = debug_enabled;
        mixer->has_prev_mix_frame = false;

        int error = 0;


        mixer->opus_enc_main = opus_encoder_create(MIX_AUIDO_FS, MIX_AUDIO_CHANNELS, MIX_AUDIO_OPUS_APP, &error);
        opus_encoder_ctl(mixer->opus_enc_main, OPUS_SET_COMPLEXITY_REQUEST, 8);
        opus_encoder_ctl(mixer->opus_enc_main, OPUS_SET_INBAND_FEC_REQUEST, 1);
        opus_encoder_ctl(mixer->opus_enc_main, OPUS_SET_PACKET_LOSS_PERC_REQUEST, 20);
        int opus_lsb = app_config_get()->mixer_opus_lsb;
        if (opus_lsb >= 8 && opus_lsb < 16) {
            opus_encoder_ctl(mixer->opus_enc_main, OPUS_SET_LSB_DEPTH(opus_lsb));
            opus_encoder_ctl(mixer->opus_enc_main, OPUS_SET_DTX(1));
        }
        if (debug_enabled) {
            mixer->debug = open_mixer_debug_file(mixer);
        }
        mixer->start_time = get_timestamp_ms();
        mixer->max_talkers = app_config_get()->mixer_talker_number;
        mixer->remote = rmixer_new(mixer->id, g_rmixer_callback, mixer);
        addr = rmixer_get_local_address(mixer->remote);
        ret = 0;
    } while (0);
    
    if(ret){
        xmixer_delete(mixer);
        mixer = NULL;
    }
    
    return mixer;
}

static
void on_mixer_rtc_destroy(rtc_endpoint_t rtc, void * context){
    mix_dbgi("on_mixer_rtc_destroy");
    xmixer_stream_t obj = (xmixer_stream_t) context;
    xmixer_stream_delete(obj);
    mix_dbgi("on_mixer_rtc_destroy end");
}

// todo: don't rely callback to push data, this makes unit test relys rtc and callbacks
static
bool on_mixer_rtc_stream_data(rtc_endpoint_t rtc, void * context, int stream_type,
                              rtc_codec_id_t codec, rtc_codec_id_t fec_codec, int data_type,
                              int is_rtcp, int is_rtx, unsigned char * buf, int len) {
    if (is_rtx) {
        return false;
    }
    xmixer_stream_t obj = (xmixer_stream_t) context;
    xmixer_t mixer = obj->mixer;
    if(data_type == RTC_RECV_CLEAR){
        if(stream_type == RTC_AUDIO && !is_rtcp){
            int audio_level = 0;
            int is_vad = 0;
            // todo: prepush is mainly getting ssrc-audio-level, this should be done outside
            audio_recveq_prepush(obj->recveq,  buf, len, &audio_level, &is_vad);
            if(audio_level >= 0){
                audio_level = 0-audio_level;
            }else{
                audio_level = MIX_AUDIO_MIN_LEVEL;
            }

            if(obj->talker){
                if(obj->talker->mxstream != obj){
                    dbge("mixer |%s| found inconsist talker and mxstream", mixer->id);
                }
                audio_recveq_commit_push(obj->recveq);

                // 只转发本地rtc到其他mixer
                if (!obj->talker->mxstream->is_remote && mixer->remote != nullptr) {
                    unsigned char send_buf[1600] = {0, };
                    memcpy(send_buf, buf, len);
                    rtp_set_ssrc(buf, len, obj->talker->mxstream->ssrc);
                    // 此时的pt是未替换的
                    rmixer_send_data(mixer->remote, buf, len);
                }
            }else{
                if(audio_level > mixer->candidate_audio_level){
                    //dbgi("mixer |%s| rtc |%s| candidate level %d > %d", mixer->id, rtc_endpoint_get_id(obj->rtc), audio_level, mixer->candidate_max_audio_level);
                    mixer->candidate_audio_level = audio_level;
                    mixer->candidate_stream = obj;
                }
            }

            obj->last_audio_level = audio_level;
//            obj->audio_level_sum += audio_level;
//            ++obj->num_audio_levels;
//            if(obj->num_audio_levels >= 40){
//                obj->audio_level_sum = 0;
//                obj->num_audio_levels = 0;
//            }
            // return true will avoid rtc connector from receiving audio data
            // return true;
            // commented due to allow old mobile sdk listen to publisher with subscribe
        } else if (stream_type == RTC_AUDIO && is_rtcp) {
            // 转发rtcp到其他mixer, 否则如果该rtc不说话，就听不见别的实例的别人了
            if (mixer->remote != nullptr) {
                unsigned char send_buf[1600] = {0, };
                memcpy(send_buf, buf, len);
                rmixer_send_data(mixer->remote, buf, len);
            }
        }
    }
    return false;
}

static
void on_mixer_rtc_stream_event(rtc_endpoint_t rtc, void * context, int event, void * event_arg1, int event_arg2){
    xmixer_stream_t mxstream = (xmixer_stream_t) context;
    xmixer_t mixer = mxstream->mixer;

    if(event == RTC_EVENT_SETUP){
        dbgi("mixer |%s| on rtc setup [%s]", mixer->id, mxstream->rtc_id);
        if(mixer->next_mix_time == LLONG_MAX){
            mixer->next_mix_time = get_timestamp_ms() + MIX_TIMER_INTERVAL;
            dbgi("mixed |%s| start mixing", mixer->id);
        }
    }
}

static
rtc_endpoint_callbacks g_mixer_for_rtc_listener = {
    .context = NULL, // context
    .on_destroy = on_mixer_rtc_destroy,
    .on_stream_data = on_mixer_rtc_stream_data, //  on_stream_data;
    .on_event = on_mixer_rtc_stream_event
};

int xmixer_add_rtc(xmixer_t mixer, rtc_endpoint_t rtc){
    mix_dbgi("mixer |%s| add rtc |%s|", mixer->id, rtc_endpoint_get_id(rtc));
    int ret = -1;
    xmixer_stream_t mxstream = NULL;
    do {
        ZERO_ALLOC(mxstream, xmixer_stream_t, sizeof(struct xmixer_stream_st));
        mxstream->is_remote = false;
        mxstream->ssrc = rtc_endpoint_get_remote_audio_ssrc(rtc);
        mxstream->rtc = rtc;
        mxstream->rtc_id = g_strdup(rtc_endpoint_get_id(rtc));
        mxstream->mixer = mixer;
        mxstream->last_audio_level = MIX_AUDIO_MIN_LEVEL;

        mixer->mxstreams->push_front(mxstream);
        mxstream->it_for_owner = mixer->mxstreams->begin();
        
        mxstream->recveq = audio_recveq_create();
        rtc_codecs_t* codecs = rtc_endpoint_codecs(rtc);
        rtc_codec_t* codec_opus = rtc_codecs_find_by_id(codecs, rtc_codec_id_opus);
        if (!codec_opus) {
            dbge("xmixer: cannot add rtc, opus codec not found");
            ret = -2;
            break;
        }
        audio_recveq_reg_codec(mxstream->recveq, "opus", rtc_codec_default_pt_opus);
//        RtpExtensionMap* ext = rtc_endpoint_audio_extension_map(rtc);
//        if (ext) {
//            uint8_t audio_level_extension_id = static_cast<uint8_t>(ext->getId(RtpExtensionType::SSRCAudioLevel));
//            if (audio_level_extension_id) {
//                audio_recveq_register_audio_level(mxstream->recveq, audio_level_extension_id);
//            }
//        }
        audio_recveq_register_audio_level(mxstream->recveq, RtpExtensionType::SSRCAudioLevel);

        int error = 0;
        mxstream->opus_enc = opus_encoder_create(MIX_AUIDO_FS, MIX_AUDIO_CHANNELS, MIX_AUDIO_OPUS_APP, &error);
        opus_encoder_ctl(mxstream->opus_enc, OPUS_SET_COMPLEXITY_REQUEST, 5);
        opus_encoder_ctl(mxstream->opus_enc, OPUS_SET_INBAND_FEC_REQUEST, 1);
        opus_encoder_ctl(mxstream->opus_enc, OPUS_SET_PACKET_LOSS_PERC_REQUEST, 20);
        int opus_lsb = app_config_get()->mixer_opus_lsb;
        if (opus_lsb >= 8 && opus_lsb < 16) {
            opus_encoder_ctl(mxstream->opus_enc, OPUS_SET_LSB_DEPTH(opus_lsb));
            opus_encoder_ctl(mxstream->opus_enc, OPUS_SET_DTX(1));
        }

        if (mixer->debug_enabled) {
            open_debug_input_tlv(mixer, mxstream);
            open_debug_output_tlv(mixer, mxstream);
        }
        mxstream->in_dtx = false;

        rtc_endpoint_register_callback(rtc, mxstream, &g_mixer_for_rtc_listener);
        ret = 0;
    } while (0);
    
    if(ret){
        xmixer_stream_delete(mxstream);
        mxstream = NULL;
    }
    
    return ret;
}

static inline
int pull_talker_pcm(xmixer_talker_t talker){
    uint32_t timestamp;
    int16_t * sample_data;
    int samples_per_channel;
    int num_channels;

    xmixer_stream_t mxstream = talker->mxstream;
    int16_t * pcm_buf = talker->pcm_buf;
    talker->pcm_samples = 0;
    
    int ret = 0;
    for(int np = 0; np < 2; np++){
        ret = audio_recveq_pull_pcm(mxstream->recveq, &timestamp, &sample_data, &samples_per_channel, &num_channels);
        if(ret) {
            dbge("xmixer: pull pcm fail !!!!!!!!!!!!");
            break;
        }
        int copy_bytes = samples_per_channel * num_channels * sizeof(int16_t);
        memcpy(pcm_buf+talker->pcm_samples, sample_data, copy_bytes);
        talker->pcm_samples += samples_per_channel * num_channels;
        
    } // foreach 10ms
    if(ret){
        talker->pcm_samples = 0;
    }
    return ret;
}

static inline
void clamping_pcm_samples(int32_t main_frame[], int num_samples, int16_t mix_frame[]){
    for(int i = 0; i < num_samples; i++){
        int32_t z = main_frame[i];
        if(z > 32767){
            z = 32767;
        }else if(z < -32768){
            z = -32768;
        }
        mix_frame[i] = (int16_t)z;
    }
}

static
int _xmixer_make_rtp_header(rtc_endpoint_t rtc, uint8_t* buf, uint16_t seq, uint32_t ts, std::list<csrc_audio_level>& csrcs) {
    //     0                   1                   2                   3
    //     0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
    //     +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
    //     |V=2|P|X|  CC   |M|     PT      |       sequence number         |
    //     +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
    //     |                           timestamp                           |
    //     +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
    //     |           synchronization source (SSRC) identifier            |
    //     +=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+
    //     |            contributing source (CSRC) identifiers             |
    //     |                             ....                              |
    //     +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
    uint8_t cc = 0;
    for (auto csrc : csrcs) {
        if (csrc.audio_level < MIX_AUDIO_CONTRIBUTING_LEVEL) {
            continue;
        }
        cc += 1;
        // rtp could at most have 15 csrc
        if (cc == 15) {
            break;
        }
    }

    int x = 0;
    int csrcAudioLevelId = 0;
    int ssrcAudioLevelId = 0;
    if (rtc) {
        auto ext = rtc_endpoint_audio_extension_map(rtc);
        csrcAudioLevelId = ext->getId(RtpExtensionType::CSRCAudioLevel);
        ssrcAudioLevelId = ext->getId(RtpExtensionType::SSRCAudioLevel);
        if ((csrcAudioLevelId && cc) || ssrcAudioLevelId) {
            x = 1;
        }
    }
    int offset = 0;
    buf[offset++] = 2 << 6 | 0 << 5 | x << 4 | cc;
    buf[offset++] = 1 << 7 | rtc_codec_default_pt_opus;
    be_set_u16(seq, buf + offset);
    offset += 2;
    be_set_u32(ts, buf + offset);
    offset += 4;
    be_set_u32(MIX_RTP_SSRC, buf + offset);
    offset += 4;

    // audio level ranges from -127(zero energy) to 0 (full of energy)
    int8_t max_audio_level = -127;
    int i = 0;
    for (auto csrc : csrcs) {
        if (csrc.audio_level > max_audio_level) {
            max_audio_level = csrc.audio_level;
        }
        if (csrc.audio_level < MIX_AUDIO_CONTRIBUTING_LEVEL) {
            continue;
        }
        be_set_u32(csrc.csrc, buf+12+4*i);
        offset += 4;
        ++i;
        if (i == 15) {
            break;
        }
    }

    // write audio-level only if rtp extension header is present,
    // prefer csrc-audio-level against ssrc-audio-level
    //
    //   0                   1                   2                   3
    //   0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
    //   +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
    //   |       0xBE    |    0xDE       |           length=3            |
    //   +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
    //   |  ID   | L=0   |     data      |  ID   |  L=1  |   data...
    //   +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
    //         ...data   |    0 (pad)    |    0 (pad)    |  ID   | L=3   |
    //   +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
    //   |                          data                                 |
    //   +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
    if (csrcAudioLevelId && cc) {
        buf[offset++] = 0xBE;
        buf[offset++] = 0xDE;
//            auto n = csrcs.size() - (exclude_csrc ? 1 : 0);
        auto extLen = 1 + cc;
        extLen = (extLen + 3) / 4;
        be_set_u16(extLen, buf + offset);
        offset += 2;
        //   0                   1                   2                   3
        //   0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
        //   +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
        //   |  ID   | len=2 |0|   level 1   |0|   level 2   |0|   level 3   |
        //   +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
        //
        // here we used type of the extension, which get translated by send_rtp_data
        //buf[offset++] = csrcAudioLevelId << 4 | (n - 1);
        buf[offset++] = RtpExtensionType::CSRCAudioLevel << 4 | (cc - 1);
        int i = 0;
        for (auto csrc : csrcs) {
            if (csrc.audio_level < MIX_AUDIO_CONTRIBUTING_LEVEL) {
                continue;
            }
            buf[offset++] = -csrc.audio_level | 0x80;
            i++;
            if (i == 15) {
                break;
            }
        }
    } else if (ssrcAudioLevelId){
        buf[offset++] = 0xBE;
        buf[offset++] = 0xDE;
        be_set_u16(1, buf + offset);
        offset += 2;
        //    0                   1
        //    0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5
        //    +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
        //    |  ID   | len=0 |V| level       |
        //    +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
        //
        // here we used type of the extension, which get translated by send_rtp_data
        // buf[offset++] = ssrcAudioLevelId << 4 | 0;
        buf[offset++] = RtpExtensionType::SSRCAudioLevel << 4 | 0;
        buf[offset++] = -max_audio_level | 0x80;
    }
    if (csrcAudioLevelId || ssrcAudioLevelId) {
        while (offset % 4) {
            buf[offset++] = 0;
        }
    }

    return offset;
}


static
void xmixer_mix_frame(xmixer_t mixer, int64_t now_ms){
    auto cfg = app_config_get();

    int32_t main_frame[MIX_AUDIO_SAMPLES_20MS];
    int16_t * out_frame = mixer->stream_frame;
    int16_t * mix_frame = mixer->mix_frame;

    int num_samples = MIX_AUDIO_SAMPLES_20MS;
    int num_talker = 0;
    int ret = 0;

    std::list<csrc_audio_level> csrc_levels;

    int update_talker = 0;

    memset(main_frame, 0, sizeof(main_frame));

    xmixer_talker_t mix_talkers[MIX_AUIDO_MAX_TALKER+1] = {nullptr,};
    int num_mix_talkers = 0;

    xmixer_talker_t empty_talker = NULL;
    xmixer_talker_t min_talker = NULL;
    int8_t min_level = 0;

    for(int ti = 0; ti < mixer->max_talkers; ti++){
        xmixer_talker_t talker = &mixer->talkers[ti];
        xmixer_stream_t mxstream = talker->mxstream;

        if(!mxstream){
            empty_talker = talker;
            continue;
        }

        talker->op = TALK_OP_NORMAL;

        ret = pull_talker_pcm(talker);
        if(ret == 0){
            auto level_and_energy = calculate_audio_level((int16_t *) talker->pcm_buf, talker->pcm_samples);
            int level = level_and_energy.first;
            double energy = level_and_energy.second;
            talker->audio_level = level;
            // calc audio-level average
            double avg = 0.0;
            int s = 0;
            if (talker->audio_level_hist_t > talker->audio_level_hist_h) {
                s = talker->audio_level_hist_t - talker->audio_level_hist_h;
            } else if (talker->audio_level_hist_t < talker->audio_level_hist_h) {
                s = MIX_AUDIO_HISTORY_SIZE - talker->audio_level_hist_h + talker->audio_level_hist_t;
            }
            // not full
            if (s < MIX_AUDIO_HISTORY_SIZE - 1) {
                avg = std::pow(10, (double)talker->audio_level_avg / 10.0) * s;
                talker->audio_level_hist[talker->audio_level_hist_t] = energy;
                avg += (double)energy;
                avg /= (s + 1);
                talker->audio_level_avg = 10 * log10(avg);
                if (talker->audio_level_avg < -127) {
                    talker->audio_level_avg = -127;
                } else if (talker->audio_level_avg > 0) {
                    talker->audio_level_avg = 0;
                }
                talker->audio_level_hist_t += 1;
                if (talker->audio_level_hist_t >= MIX_AUDIO_HISTORY_SIZE) {
                    talker->audio_level_hist_t = 0;
                }
                if (talker->audio_level_hist_t == talker->audio_level_hist_h) {
                    talker->audio_level_hist_h += 1;
                    if (talker->audio_level_hist_h >= MIX_AUDIO_HISTORY_SIZE) {
                        talker->audio_level_hist_h = 0;
                    }
                }
                //std::clog << "audio history " << talker->audio_level_hist_h << " " << talker->audio_level_hist_t << " " << level << " " << talker->audio_level_avg << std::endl;
            } else {
                talker->audio_level_hist[talker->audio_level_hist_t] = energy;
                avg = std::pow(10, (double)talker->audio_level_avg / 10.0);
                double head = talker->audio_level_hist[talker->audio_level_hist_h];
                avg -= (double)head / s;
                avg += (double)energy / s;
                talker->audio_level_avg = 10 * log10(avg);
                if (talker->audio_level_avg < -127) {
                    talker->audio_level_avg = -127;
                } else if (talker->audio_level_avg > 0) {
                    talker->audio_level_avg = 0;
                }

                // talker->audio_level_history.pop_front();
                talker->audio_level_hist_h += 1;
                if (talker->audio_level_hist_h >= MIX_AUDIO_HISTORY_SIZE) {
                    talker->audio_level_hist_h = 0;
                }
                // talker->audio_level_history.push_back(level);
                talker->audio_level_hist_t += 1;
                if (talker->audio_level_hist_t >= MIX_AUDIO_HISTORY_SIZE) {
                    talker->audio_level_hist_t = 0;
                }
                //std::clog << "audio history2 " << talker->audio_level_hist_h << " " << talker->audio_level_hist_t << " - " << head << " + "<< level << " = " << talker->audio_level_avg << std::endl;
            }

            // check talker's audio level lower than MIX_AUDIO_CONTRIBUTING_LEVEL
            if (talker->audio_level < MIX_AUDIO_CONTRIBUTING_LEVEL) {
                if (talker->silence_start_time) {
                    int64_t silence_duration = now_ms - talker->silence_start_time;
                    // clear talker if lower than MIX_AUDIO_CONTRIBUTING_LEVEL for more than MIX_AUDIO_TALKER_SILENCE_PERIOD
                    if (silence_duration >= MIX_AUDIO_TALKER_SILENCE_PERIOD) {
                        if (cfg->enable_mixer_debug) {
                            dbgi("mixer |%s| talker |%s| silenced %" PRId64 " - %" PRId64 " = %" PRId64 "ms", mixer->id, rtc_endpoint_get_id(talker->mxstream->rtc), now_ms, talker->silence_start_time, silence_duration);
                        }
                        talker->silence_start_time = 0;
                        audio_recveq_flush(talker->mxstream->recveq);
                        unbind_stream_from_talker(talker);
                        empty_talker = talker;
                        continue;
                    }
                } else {
                    // set start of silence
                    talker->silence_start_time = now_ms;
                }
            } else {
                // clear start of silence
                talker->silence_start_time = 0;
            }

            if(talker->audio_level_avg < min_level){
                min_talker = talker;
                min_level = talker->audio_level_avg;
            }
            mix_talkers[num_mix_talkers] = talker;
            ++num_mix_talkers;
        }else{
            talker->pcm_samples = 0;
        }
    } // for talker

    mixer->passive_talker.op = TALK_OP_NONE;
    if(mixer->passive_talker.mxstream){
        ret = pull_talker_pcm(&mixer->passive_talker);
        if (ret) {
            dbge("could not pull passive talker!!!");
        }
        auto level_and_energy = calculate_audio_level((int16_t *) mixer->passive_talker.pcm_buf,
                                                                  mixer->passive_talker.pcm_samples);
        mixer->passive_talker.audio_level = level_and_energy.first;
        if(empty_talker || (min_talker && mixer->passive_talker.audio_level > min_level && mixer->passive_talker.audio_level > MIX_AUDIO_CONTRIBUTING_LEVEL)){
            if(!empty_talker){
                min_talker->op = TALK_OP_RAMPOUT;
            }
            mixer->passive_talker.op = TALK_OP_RAMPIN;
            mix_talkers[num_mix_talkers] = &mixer->passive_talker;
            ++num_mix_talkers;
        } else {
            mixer->passive_talker.op = TALK_OP_NONE;
            mixer->passive_talker.pcm_samples = 0;
        }
    }

    for(int ti = 0; ti < num_mix_talkers; ti++){
        xmixer_talker_t talker = mix_talkers[ti];

        if(talker->op == TALK_OP_RAMPIN){
            rampin_talker_pcm(talker);
            csrc_levels.emplace_back(csrc_audio_level{talker->mxstream->ssrc, talker->audio_level});
        }else if(talker->op == TALK_OP_RAMPOUT){
            rampout_talker_pcm(talker);
        } else {
            csrc_levels.emplace_back(csrc_audio_level{talker->mxstream->ssrc, talker->audio_level});
        }

        for(int i = 0; i < talker->pcm_samples; i++){
            main_frame[i] += talker->pcm_buf[i];
        }
        ++num_talker;
    } // for streams

    ++mixer->num_mixed_frames;
    uint32_t rtp_timestamp = mixer->num_mixed_frames * (48000 / 1000 * 20); // TODO: change sample rate

    int main_encoded_len = 0;
    int main_rtp_header_len = _xmixer_make_rtp_header(nullptr, mixer->mixed_encoded_buf, mixer->recording_rtp_seq++, rtp_timestamp, csrc_levels);

    // copy last mix frame before writing to it, the last frame is used to initialize newly used encoder
    memcpy(mixer->prev_mix_frame, mixer->mix_frame, sizeof(int16_t) * MIX_AUDIO_SAMPLES_20MS);
    mixer->has_prev_mix_frame = true;

    clamping_pcm_samples(main_frame, num_samples, mix_frame);

    ret = opus_encode(mixer->opus_enc_main, mix_frame, MIX_AUDIO_SAMPLES_20MS/MIX_AUDIO_CHANNELS, mixer->mixed_encoded_buf+main_rtp_header_len, sizeof(mixer->mixed_encoded_buf)-main_rtp_header_len);
    // main_encoded_len does not include rtp header 12 bytes
    main_encoded_len = ret;
    // recording
    if (mixer->recording_tlv) {
        int64_t ts = get_timestamp_ms();
        uint32_t ts_hi = (uint32_t)((ts>>32) & 0xFFFFFFFFUL);
        uint32_t ts_lo = (uint32_t)(ts & 0xFFFFFFFFUL);

        uint8_t ts_buf[8];
        be_set_u32(ts_hi, ts_buf+0);
        be_set_u32(ts_lo, ts_buf+4);

        tlv_file_write2(mixer->recording_tlv, TLV_TYPE_RTP, 8, ts_buf, main_rtp_header_len+main_encoded_len, mixer->mixed_encoded_buf);
    }

    int encoder_id = 0;
    for (std::list<xmixer_stream_t>::iterator it = mixer->mxstreams->begin(); it != mixer->mxstreams->end(); ++it) {
        xmixer_stream_t mxstream = *it;
        // 来自remote mixer的stream，没有rtc，不需要发送RTP
        if (mxstream->is_remote) {
            continue;
        }
        encoder_id += 1;

        uint8_t * encoded_buf = NULL;
        int encoded_len = 0;

        if(mixer->is_loopback || !mxstream->talker || !mxstream->talker->pcm_samples){
            if (main_encoded_len <= 2) {
                if (mxstream->in_dtx) {
                    continue;
                }
                mxstream->in_dtx = true;
            } else {
                mxstream->in_dtx = false;
            }
            // note: use mxstream->encoded_buf so that we will not overwrite opus encoded data
            // since this rtp header usually larger than main rtp header
            int rtp_header_len = _xmixer_make_rtp_header(mxstream->rtc, mxstream->encoded_buf, mxstream->rtp_seq++, rtp_timestamp, csrc_levels);
            memcpy(mxstream->encoded_buf+rtp_header_len, mixer->mixed_encoded_buf + main_rtp_header_len, main_encoded_len);
            encoded_buf = mxstream->encoded_buf;
            encoded_len = rtp_header_len + main_encoded_len;
            if (mxstream->debug_input_tlv) {
                int16_t zeros[MIX_AUDIO_SAMPLES_20MS] = {0,};
                write_debug_input_tlv(mxstream, 0, (uint8_t*)zeros, num_samples * 2);
            }
            // write to debug recording
            if (mxstream->debug_output_tlv) {
                write_debug_output_tlv(mxstream, 0, (uint8_t*)mix_frame, num_samples * 2);
            }
        }else{
            if (mixer->debug) {
                write_mixer_debug_file(mixer->debug, mxstream->rtc, mxstream->talker->pcm_buf, 16000 / 1000 * 20);
            }
            if (mxstream->debug_input_tlv) {
                write_debug_input_tlv(mxstream, encoder_id, (uint8_t*)mxstream->talker->pcm_buf, num_samples * 2);
            }
            int talker_rtp_header_len = _xmixer_make_rtp_header(mxstream->rtc, mxstream->encoded_buf, mxstream->rtp_seq++, rtp_timestamp, csrc_levels);
            int16_t* talker_frame;
            // initialize encoder if it was not used in last round
            if (mxstream->talker->op == TALK_OP_RAMPIN) {
                if (mixer->has_prev_mix_frame) {
                    opus_encode(mxstream->opus_enc, mixer->prev_mix_frame, MIX_AUDIO_SAMPLES_20MS/MIX_AUDIO_CHANNELS, mxstream->encoded_buf+talker_rtp_header_len, sizeof(mxstream->encoded_buf)-talker_rtp_header_len);
                }
            }

            for(int i = 0; i < mxstream->talker->pcm_samples; i++){
                int32_t z = main_frame[i];
                z -= (int32_t) mxstream->talker->pcm_buf[i];
                if(z > 32767){
                    z = 32767;
                }else if(z < -32768){
                    z = -32768;
                }
                out_frame[i] = (int16_t)z;
            }
            talker_frame = out_frame;

            mxstream->talker->pcm_samples = 0;
            ret = opus_encode(mxstream->opus_enc, talker_frame, MIX_AUDIO_SAMPLES_20MS/MIX_AUDIO_CHANNELS, mxstream->encoded_buf+talker_rtp_header_len, sizeof(mxstream->encoded_buf)-talker_rtp_header_len);
            if (ret < 0) {
//                if (ret != mxstream->opus_err) {
//                    mix_dbge("mixer |%s| could not encode for |%s| : %s", mixer->id, rtc_endpoint_get_id(mxstream->rtc), opus_strerror(ret));
//                    mxstream->opus_err = ret;
//                }
                continue;
            } else if (ret <= 2) {
                if (mxstream->in_dtx) {
                    mxstream->rtp_seq -= 1;
                    continue;
                }
                mxstream->in_dtx = true;
            } else {
                mxstream->in_dtx = false;
            }

            encoded_buf = mxstream->encoded_buf;
            encoded_len = talker_rtp_header_len+ret;


            // write to debug output
            if (mxstream->debug_output_tlv) {
                write_debug_output_tlv(mxstream, encoder_id, (uint8_t*)talker_frame, num_samples*2);
            }
        }

        // send encoded data to rtp
        if(encoded_len > 0){
            rtc_endpoint_send_stream_data(mxstream->rtc, RTC_AUDIO, rtc_codec_id_opus, rtc_codec_id_unknown, 0, 0, encoded_buf, encoded_len);
        }

    }

    if (mixer->debug) {
        flush_mixer_debug_file(mixer->debug);
    }

    if(mixer->passive_talker.op == TALK_OP_RAMPIN){
        if(empty_talker){
            // passive talker -> empty active talker
            if (cfg && cfg->enable_mixer_debug) {
                mix_dbgi("mixer |%s| updated: passive talker -> empty active talker, [%s]", mixer->id, mixer->passive_talker.mxstream->rtc_id);
            }
            xmixer_stream_t stream1 = mixer->passive_talker.mxstream;
            unbind_stream_from_talker(&mixer->passive_talker);
            bind_stream_with_talker(empty_talker, stream1);
            empty_talker = NULL;
            update_talker = 1;
        }else{
            // active talker <-> passive talker
            // clear silence start time to avoid false silence
            min_talker->silence_start_time = 0;
            // clear audio level history
            //min_talker->audio_level_avg = 0.0;
            min_talker->audio_level_hist_h = 0;
            min_talker->audio_level_hist_t = 0;

            audio_recveq_flush(min_talker->mxstream->recveq);

            xmixer_talker_t talker1 = min_talker;
            xmixer_talker_t talker2 = &mixer->passive_talker;

            xmixer_stream_t stream1 = talker1->mxstream;
            xmixer_stream_t stream2 = talker2->mxstream;

            unbind_stream_from_talker(talker1);
            unbind_stream_from_talker(talker2);

            bind_stream_with_talker(talker1, stream2);
            bind_stream_with_talker(talker2, stream1);



            update_talker = 1;
            if (cfg && cfg->enable_mixer_debug) {
                mix_dbgi("mixer |%s| updated: active talker <-> passive talker, [%s]<->[%s]", mixer->id,
                         stream1->rtc_id, stream2->rtc_id);
            }
        }

    }


    if(mixer->candidate_stream){
        if(empty_talker){
            // candidate stream -> empty active talker
            if (cfg && cfg->enable_mixer_debug) {
                mix_dbgi("mixer |%s| updated: candidate stream -> empty active talker, [%s]",
                         mixer->id,
                         mixer->candidate_stream->rtc_id);
            }
            bind_stream_with_talker(empty_talker, mixer->candidate_stream);
            mixer->candidate_stream = NULL;
            update_talker = 1;
        }else if(!mixer->passive_talker.mxstream){
            // candidate stream -> empty passive talker
            if (cfg && cfg->enable_mixer_debug) {
                mix_dbgi("mixer |%s| updated: candidate stream -> empty passive talker, [%s]",
                         mixer->id,
                         mixer->candidate_stream->rtc_id);
            }
            bind_stream_with_talker(&mixer->passive_talker, mixer->candidate_stream);
            mixer->candidate_stream = NULL;
            update_talker = 1;
        }else if(mixer->passive_talker.mxstream->last_audio_level < mixer->candidate_stream->last_audio_level){
            // candidate stream -> passive talker
            if (cfg && cfg->enable_mixer_debug) {
                mix_dbgi("mixer |%s| updated: candidate stream -> passive talker, [%s]->[%s]", mixer->id,
                         mixer->candidate_stream->rtc_id,
                         mixer->passive_talker.mxstream->rtc_id);
            }
            xmixer_stream_t old_stream = mixer->passive_talker.mxstream;
            unbind_stream_from_talker(&mixer->passive_talker);
            bind_stream_with_talker(&mixer->passive_talker, mixer->candidate_stream);
            mixer->candidate_stream = NULL;
            audio_recveq_flush(old_stream->recveq);
            update_talker = 1;
        }

    }

    if(update_talker){
        ++mixer->talker_update_seq;
        if (cfg && cfg->enable_mixer_debug) {
            dump_talkers(mixer);
        }
    }

    if (now_ms - mixer->last_dump_time > 10000) {
        dump_talkers(mixer);
        mixer->last_dump_time = now_ms;
    }

    mixer->candidate_stream = NULL;
    mixer->candidate_audio_level = MIX_AUDIO_RESET_LEVEL;
}



bool xmixer_maybe_mix_frame(xmixer_t mixer){
    int64_t now_ts = get_timestamp_ms();

    if (mixer->remote != nullptr) {
        rmixer_handle_timeout(mixer->remote, now_ts);
    }

    if(mixer->mxstreams->size() == 0){
        return false;
    }

    while(now_ts > mixer->next_mix_time){
        //        dbgi("catching up timer");
        xmixer_mix_frame(mixer, now_ts);
        mixer->next_mix_time += MIX_TIMER_INTERVAL;
        now_ts = get_timestamp_ms();
        //        dbgi("timer diff %lld", (now_ts-mixer->next_mix_time));
    }


    return true;
}

int xmixer_set_tlv_filename(xmixer_t mixer, const std::string& tlv_pathname){
    if(tlv_pathname.empty()){
        return 0;
    }

    if(mixer->recording_tlv_pathname){
        dbge("mixer set tlv filename %s failed : already set to %s", tlv_pathname.c_str(), mixer->recording_tlv_pathname);
        return -1;
    }
    mixer->recording_tlv_pathname = strdup(tlv_pathname.c_str());
    std::string tlv_path = tlv_pathname;
    if (tlv_path.back() == '/') {
        tlv_path.substr(0, tlv_path.length()-1);
    }
    auto last_sep = tlv_path.rfind('/');
    tlv_path = tlv_path.substr(0, last_sep);
    if (tlv_path.length()) {
        mkdir_tree(tlv_path.c_str());
    }
    mixer->recording_tlv = tlv_file_open(tlv_pathname.c_str());
    if (!mixer->recording_tlv) {
        dbge("mixer |%s| could not open tlv file %s", mixer->id, tlv_pathname.c_str());
        return -1;
    }
    dbgi("mixer |%s| opened recording tlv %s", mixer->id, tlv_pathname.c_str());
    uint8_t codec_buf[128];

    auto codec_name = rtc_codec_name(rtc_codec_id_opus);
    auto codec_name_len = strlen(codec_name) + 1;
    auto codec_type = rtc_codec_type(rtc_codec_id_opus);
    be_set_u32(codec_type, codec_buf + 0);
    be_set_u32(MIX_RTP_PT,  codec_buf + 4);
    be_set_u32(MIX_AUIDO_FS, codec_buf + 8);
    be_set_u32(MIX_AUDIO_CHANNELS, codec_buf + 12);
    be_set_u32(codec_name_len, codec_buf + 16);
    memcpy(codec_buf + 20, codec_name, codec_name_len);
    int err = tlv_file_write(mixer->recording_tlv, TLV_TYPE_CODEC, 20 + codec_name_len, codec_buf);
    if (err) {
        dbge("mixer |%s| could not write tlv-codec %s", mixer->id, codec_name);
        tlv_file_close(mixer->recording_tlv);
        mixer->recording_tlv = nullptr;
        return -1;
    }

    return 0;
}

void xmixer_get_tlv_filename(xmixer_t mixer, std::string& tlv_pathname) {
    tlv_pathname = "";
    if (mixer->recording_tlv_pathname) {
        tlv_pathname = std::string(mixer->recording_tlv_pathname);
    }
}

const char *xmixer_get_id(xmixer_t mixer) {
    return mixer->id;
}

static void xmixer_handle_remote_stream_new(void* data, uint32_t ssrc) {
    auto mixer = static_cast<xmixer_t>(data);

    audio_recveq_t recveq = audio_recveq_create();
    audio_recveq_reg_codec(recveq, "opus", rtc_codec_default_pt_opus);
    audio_recveq_register_audio_level(recveq, RtpExtensionType::SSRCAudioLevel);

    int err = 0;
    auto encoder = opus_encoder_create(MIX_AUIDO_FS, MIX_AUDIO_CHANNELS, MIX_AUDIO_OPUS_APP, &err);
    opus_encoder_ctl(encoder, OPUS_SET_COMPLEXITY_REQUEST, 5);
    opus_encoder_ctl(encoder, OPUS_SET_INBAND_FEC_REQUEST, 1);
    opus_encoder_ctl(encoder, OPUS_SET_PACKET_LOSS_PERC_REQUEST, 20);

    auto stream = new xmixer_stream_st();
    stream->is_remote = true;
    stream->ssrc = ssrc;
    stream->rtc = nullptr;
    stream->rtc_id = g_strdup("remote");
    stream->mixer = mixer;
    stream->last_audio_level = MIX_AUDIO_MIN_LEVEL;
    stream->recveq = recveq;
    stream->in_dtx = false;
    stream->audio_level_sum = 0;
    stream->num_audio_levels = 0;
    stream->opus_enc = encoder;

    mixer->streams.emplace(ssrc, stream);

}

static void xmixer_handle_remote_stream_data(void* data, uint32_t ssrc, const unsigned char* buf, int size) {
    auto mixer = static_cast<xmixer_t>(data);

    auto it = mixer->streams.find(ssrc);

    if (it == mixer->streams.end()) {
        return;
    }

    auto stream = it->second;
    int audio_level = 0;
    int is_active = 0;

    // todo: audio_recveq_prepush只是读取audio level，没必要放在recveq内，还得注册rtp extension
    // 而且读出来的audio_level还需要再处理一下
    // int ret = audio_recveq_prepush(stream->recveq, buf, size, &audio_level, &is_active);
    audio_recveq_prepush(stream->recveq, buf, size, &audio_level, &is_active);
    if (audio_level >= 0) {
        audio_level = 0 - audio_level;
    } else {
        audio_level = MIX_AUDIO_MIN_LEVEL;
    }

    if (stream->talker) {
        audio_recveq_commit_push(stream->recveq);
    } else {
        if (audio_level > mixer->candidate_audio_level) {
            mixer->candidate_audio_level = audio_level;
            mixer->candidate_stream = stream;
        }
    }

    stream->last_audio_level = audio_level;
    return;
}

static void xmixer_handle_remote_stream_destroy(void* data, uint32_t ssrc) {
    auto mixer = static_cast<xmixer_t>(data);
    mixer->streams.erase(ssrc);
}

int xmixer_join_remote(xmixer_t mixer, sockaddr_in sa) {

    rmixer_join(mixer->remote, sa);

    return 0;
}

//int open_debug_output_ffmpeg(xmixer_stream_t mxstream) {
//    rtc_endpoint_t rtc = mxstream->rtc;
//    char fn[1024] = {0,};
//    sprintf(fn, "/tmp/mixer-out-%s.mkv", rtc_endpoint_get_id(rtc));
//    int ret = avformat_alloc_output_context2(&mxstream->debug_output_ctx, nullptr, "matroska", fn);
//    if (ret != 0) {
//        mix_dbge("mixer could not allocate debug output context: %s",  av_err2str(ret));
//        return -1;
//    }
//    AVCodec* codec = avcodec_find_encoder_by_name("libopus");
//    if (!codec) {
//        mix_dbge("mixer could not find libopus codec: %s", av_err2str(ret));
//        return -2;
//    }
//
//    mxstream->debug_output_st = avformat_new_stream(mxstream->debug_output_ctx, codec);
//    if (!mxstream->debug_output_st) {
//        mix_dbge("mixer could not create debug output stream: %s", av_err2str(ret));
//        avformat_free_context(mxstream->debug_output_ctx);
//        mxstream->debug_output_ctx = nullptr;
//        return -3;
//    }
//
//    mxstream->debug_output_st->id = 0;
//    mxstream->debug_output_st->codecpar->codec_id = codec->id;
//    mxstream->debug_output_st->codecpar->codec_type = codec->type;
//    mxstream->debug_output_st->codecpar->sample_rate = MIX_AUIDO_FS;
//    mxstream->debug_output_st->codecpar->channel_layout = AV_CH_LAYOUT_STEREO;
//    mxstream->debug_output_st->codecpar->channels = 2;
//    mxstream->debug_output_st->codecpar->format = AV_SAMPLE_FMT_S16;
//
//    mxstream->debug_output_st->time_base = AVRational {1, MIX_AUIDO_FS};
//
//    av_dump_format(mxstream->debug_output_ctx, 0, fn, 1);
//
//    ret = avio_open(&mxstream->debug_output_ctx->pb, fn, AVIO_FLAG_WRITE);
//    if (ret < 0) {
//        mix_dbge("mixer could not open debug file %s: %s", fn, av_err2str(ret));
//        avformat_free_context(mxstream->debug_output_ctx);
//        mxstream->debug_output_ctx = nullptr;
//        return -4;
//    }
//    ret = avformat_write_header(mxstream->debug_output_ctx, nullptr);
//    if (ret < 0) {
//        mix_dbge("mixer could not write header to %s: %s", fn, av_err2str(ret));
//        avio_close(mxstream->debug_output_ctx->pb);
//        avformat_free_context(mxstream->debug_output_ctx);
//        mxstream->debug_output_ctx = nullptr;
//        return -5;
//    }
//    return 0;
//}
//
//int close_debug_output_ffmpeg(xmixer_stream_t mxstream) {
//    if (mxstream->debug_output_ctx) {
//        av_write_trailer(mxstream->debug_output_ctx);
//        avformat_free_context(mxstream->debug_output_ctx);
//        mxstream->debug_output_ctx = nullptr;
//    }
//}
//
//int write_debug_output_ffmpeg(xmixer_stream_t mxstream, int encoder_id, uint8_t* buf, size_t len) {
//    if (mxstream->debug_output_ctx && len) {
//        AVPacket packet;
//        av_init_packet(&packet);
//        packet.data = buf;
//        packet.size = len;
//        packet.pts = mxstream->debug_output_ts;
//        packet.dts = packet.pts;
//        packet.stream_index = mxstream->debug_output_st->index;
//        av_interleaved_write_frame(mxstream->debug_output_ctx, &packet);
//        //av_packet_free(&packet);
//        mxstream->debug_output_ts += 20;
//    }
//}

// STREAM DEBUG

#include "xtlv_file.h"

static const int TLV_TYPE_MIXER_OPUS_MAIN = 3000;
static const int TLV_TYPE_MIXER_TALKER = 3005;

int open_debug_output_tlv(xmixer_t mixer, xmixer_stream_t mxstream) {
    char fn[5120] = {0};
    sprintf(fn, "%s/mixer-debug-out-%s.tlv", app_config_get()->tlv_path, mxstream->rtc_id);
    mxstream->debug_output_tlv = tlv_file_open(fn);
    if (mxstream->debug_output_tlv) {
        mix_dbgi("mixer |%s| opened debug tlv %s", mixer->id, fn);
        return -1;
    }
    mix_dbge("mixer |%s| could not open debug tlv %s", mixer->id, fn);
    return 0;
}

int close_debug_output_tlv(xmixer_stream_t mxstream) {
    if (mxstream->debug_output_tlv) {
        tlv_file_close(mxstream->debug_output_tlv);
        mxstream->debug_output_tlv = nullptr;
    }
    return 0;
}

int write_debug_output_tlv(xmixer_stream_t mxstream, int encoder_id, uint8_t* buf, size_t len) {
    if (mxstream->debug_output_tlv) {
        tlv_file_write(mxstream->debug_output_tlv, TLV_TYPE_MIXER_OPUS_MAIN + encoder_id, len, buf);
    }
    return 0;
}

int open_debug_input_tlv(xmixer_t mixer, xmixer_stream_t mxstream) {
    char fn[5120] = {0};
    sprintf(fn, "%s/mixer-debug-in-%s.tlv", app_config_get()->tlv_path, mxstream->rtc_id);
    mxstream->debug_input_tlv = tlv_file_open(fn);
    if (mxstream->debug_input_tlv) {
        mix_dbgi("mixer |%s| opened debug tlv %s", mixer->id, fn);
        return -1;
    }
    mix_dbge("mixer |%s| could not open debug tlv %s", mixer->id, fn);
    return 0;
}

int close_debug_input_tlv(xmixer_stream_t mxstream) {
    if (mxstream->debug_input_tlv) {
        tlv_file_close(mxstream->debug_input_tlv);
        mxstream->debug_input_tlv = nullptr;
    }
    return 0;
}

int write_debug_input_tlv(xmixer_stream_t mxstream, int encoder_id, uint8_t* buf, size_t len) {
    if (mxstream->debug_input_tlv) {
        tlv_file_write(mxstream->debug_input_tlv, TLV_TYPE_MIXER_OPUS_MAIN + encoder_id, len, buf);
    }
    return 0;
}

// MIXER DEBUG

extern "C" {
#include <libavformat/avformat.h>
#include <libavcodec/avcodec.h>
}

class mixer_debug {
public:
    std::map<rtc_endpoint_t, int> channels;
    int16_t channel_samples[16000/1000*20*8];
    AVFormatContext* output_ctx;
};

mixer_debug *open_mixer_debug_file(xmixer_t mixer) {
    char fn[5120] = {0};
    sprintf(fn, "%s/mixer-%s-debug.wav", app_config_get()->tlv_path, mixer->id);
    auto debug = new mixer_debug();
    int ret = avformat_alloc_output_context2(&debug->output_ctx, nullptr, "wav", fn);
    if (ret != 0) {
        char err[AV_ERROR_MAX_STRING_SIZE];
        av_make_error_string(err, sizeof(err), ret);
        mix_dbge("could not alloc debug output context: %s", err);
        delete debug;
        return nullptr;
    }
    AVCodec* codec = avcodec_find_encoder_by_name("pcm_s16le");
    AVStream* stream = avformat_new_stream(debug->output_ctx, codec);
    AVCodecParameters* cpar = stream->codecpar;
    cpar->format = AV_SAMPLE_FMT_S16;
    cpar->channels = 8;
    cpar->channel_layout = av_get_default_channel_layout(8);
    cpar->sample_rate = MIX_AUIDO_FS;
    cpar->codec_id = codec->id;
    cpar->codec_type = codec->type;

    ret = avio_open(&debug->output_ctx->pb, fn, AVIO_FLAG_WRITE);
    if (ret < 0) {
        char err[AV_ERROR_MAX_STRING_SIZE];
        av_make_error_string(err, sizeof(err), ret);
        mix_dbge("could not open debug output file %s: %s", fn, err);
        avformat_free_context(debug->output_ctx);
        delete debug;
        return nullptr;
    }

    ret = avformat_write_header(debug->output_ctx, nullptr);
    if (ret < 0) {
        char err[AV_ERROR_MAX_STRING_SIZE];
        av_make_error_string(err, sizeof(err), ret);
        mix_dbge("could not write to output file %s: %s", fn, err);
        avio_close(debug->output_ctx->pb);
        debug->output_ctx->pb = nullptr;
        avformat_free_context(debug->output_ctx);
        delete debug;
        return nullptr;
    }

    mix_dbgi("mixer |%s| opened debug output %s", mixer->id, fn);
    return debug;
}

void close_mixer_debug_file(mixer_debug *debug) {
    if (debug->output_ctx) {
        if (debug->output_ctx->pb) {
            av_write_trailer(debug->output_ctx);
            avio_close(debug->output_ctx->pb);
            debug->output_ctx->pb = nullptr;
        }
        avformat_free_context(debug->output_ctx);
    }
    delete debug;
}

void write_mixer_debug_file(mixer_debug *debug, rtc_endpoint_t rtc, int16_t *samples, int samples_nb) {
    int channel_id = -1;
    auto it = debug->channels.find(rtc);
    if (it == debug->channels.end()) {
        if (debug->channels.size() >= 8) {
            // we can handle 8 channels at most
            return;
        }
        channel_id = debug->channels.size();
        debug->channels.emplace(rtc, channel_id);
    } else {
        channel_id = it->second;
    }
    for (int i = 0; i < samples_nb; ++i) {
        // samples has 2 channels, save the first one
        debug->channel_samples[8 * i +channel_id] = samples[i * 2];
    }
}

void flush_mixer_debug_file(mixer_debug *debug) {
    AVPacket packet;
    av_init_packet(&packet);
    packet.data = (uint8_t*)debug->channel_samples;
    packet.size = 16000 / 1000 * 20 * 8 * 2;
    packet.stream_index = 0;
    int ret = av_interleaved_write_frame(debug->output_ctx, &packet);
    if (ret < 0) {
        char err[AV_ERROR_MAX_STRING_SIZE];
        av_make_error_string(err, sizeof(err), ret);
        mix_dbge("could not write to output file: %s", err);
    }
    memset(debug->channel_samples, 0, 16000 / 1000 * 20 * 8 * 2);
}
